package org.niit.service

import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.Answer
import org.niit.common.TService

/**
 * Date:2025/6/11
 * Author：Ys
 * Description:
 */
class EDUDataService extends TService[DStream[Answer]]{

  override def dataAnalysis(data: DStream[Answer]): Unit = {
      hotQuestionTop10(data)
      gradeQuestionTop10(data)
      hotQuestionWithSubjectTop10(data)
     studentMinScoreWithQuestionTop10(data)
     subjectWithQuestionTop10(data)
     studentScoreWithQuestionTop10(data)
  }
  //需求一：实时统计Top10热点题   --->某题出现的次数
  def hotQuestionTop10(data: DStream[Answer]): Unit = {
    // key：题目id                      value: 次数  默认 1
    data.map( line=>{
      (line.question_id,1)
    } )
      .reduceByKey( _  + _ )
      .foreachRDD(rdd=>{
        val top10: Array[(String, Int)] = rdd.sortBy(_._2, false).take(10)

      //操作数据库的代码自行编写
      println("------实时统计Top10热点题--------")
      top10.foreach(println)
    })
  }
  //需求二：实时统计答题最活跃的Top10年级  ---> 某年级出现次数最多
  def gradeQuestionTop10(data: DStream[Answer]): Unit = {
      // key：年级ID                      value: 1

    data.map( line=>{
      (line.grade_id,1)

    }).reduceByKey(_ + _)
      .foreachRDD(rdd=>{
        val top10: Array[(String, Int)] = rdd.sortBy(_._2, false).take(10)
        println("------实时统计答题最活跃的Top10年级--------")
        top10.foreach(println)
      })

  }
  //需求三：实时统计每个科目的Top10热点题   --> 某个科目中题 出现的次数
  def hotQuestionWithSubjectTop10(data: DStream[Answer]): Unit = {

    data.map( line=>{

      ( (line.subject_id,line.question_id),1 )

    }).reduceByKey( _ + _)
      .foreachRDD(rdd=>{

        val top10: Array[((String, String), Int)] = rdd.sortBy(_._2, false).take(10)
        println("------实时统计每个科目的Top10热点题--------")
        top10.foreach(println)
      })
  }
  //需求四：实时统计每位学生得分最低的题目Top10  --> 某个学生某个题最低分数
  def studentMinScoreWithQuestionTop10(data: DStream[Answer]): Unit = {
    // key : 学生ID 题目iD  value : 分数
    data.map( line=>{

      ( (line.student_id,line.question_id),line.score )

    }).foreachRDD(rdd=>{
      val top10: Array[((String, String), Int)] = rdd.sortBy(_._2, true).take(10)
      println("------实时统计每位学生得分最低的题目Top10--------")
      top10.foreach(println)
    })

  }
  //需求五：实时统计每个科目的Top10热点题 --> 某个科目中 出现次数最多的 前10道题
  def subjectWithQuestionTop10(data: DStream[Answer]): Unit = {
    //1. key : 科目ID 题目ID  value : 次数

    val reduceDS: DStream[((String, String), Int)] = data.map(line => {
      ((line.subject_id, line.question_id), 1)

    }).reduceByKey(_ + _)
      //(  (语文，选择题第一次)，200 ) ===》 ( 语文，[(选择题第一题，200)，(填空题第三题，180)] )

//    reduceDS.map(line=>{
//
//      (line._1._1,(line._1._2,line._2))
//
//    })

    val groupDS: DStream[(String, Iterable[(String, Int)])] = reduceDS.map {

      case ((subjectId, questionId), count) => {

        (subjectId, (questionId, count))
      }
    }.groupByKey()

    groupDS.foreachRDD(rdd=>{
      val res = rdd.mapValues(iter=>{
        val top10: List[(String, Int)] = iter.toList.sortBy(_._2).reverse.take(10)
        top10
      })
      println("------实时统计每个科目的Top10热点题--------")
      res.foreach(println)
    })

  }
  //需求六：实时统计学生得分最低的前10道题，并带上所属题目
  def studentScoreWithQuestionTop10(data: DStream[Answer]): Unit = {
    // key: 学生id   value:题目id，分数
    val groupDS: DStream[(String, Iterable[(String, Int)])] = data.map(line => {
      (line.student_id, (line.question_id, line.score))

    }).groupByKey()

    groupDS.foreachRDD(rdd=>{
      val top10 = rdd.mapValues(iter=>{
        iter.toList.sortBy(_._2).take(10)
      })
      println("------实时统计学生得分最低的前10道题，并带上所属题目--------")
      top10.foreach(println)
    })


  }

}
